# 第二阶段步骤 10：T015 - 智能转接机制 完整执行计划

> 版本：v3.0（配置化优化版）  
> 编制时间：2026-04-29 15:40  
> 关联问题：P015（无法自动转接运营人员）  
> 预计总工时：3.5 小时

---

## 一、执行目标

### 1.1 核心目标

实现智能转接机制，解决以下两类场景：

| 场景 | 触发条件 | 转接对象 | 优先级 |
|------|----------|----------|--------|
| **场景 1** | 用户问题超出小医能力范围 | 运营人员 | P1 |
| **场景 2** | 查询到审核状态 state=1（待审核） | 全科医生团队 | P1 |

---

### 1.2 关键改进点（针对用户 3 个问题）

| 问题 | 原方案 | 优化后方案 |
|------|--------|-----------|
| **技能确认** | 假设 activity-content-audit-query 存在 | ✅ 先确认技能，再决定扩展方案 |
| **人员硬编码** | 代码中硬编码姓名/手机号 | ✅ 配置文件管理，支持动态调整 |
| **文案硬编码** | 通知模板写在代码中 | ✅ 模板配置化，运营可自主修改 |

---

## 二、前置条件确认

### 2.1 技能确认（必须）

**任务：** 确认 `activity-content-audit-query` 技能是否存在

**执行命令：**
```bash
# 方法 1：检查 workspace/skills 目录
ls -la /home/admin/.openclaw/workspace/skills/ | grep audit

# 方法 2：检查 instance/skills 目录
ls -la /home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/skills/ | grep audit

# 方法 3：全局搜索
find /home/admin/.openclaw/workspace -type d -name "activity-content-audit-query"
```

**判断标准：**
- ✅ 存在 → 在该技能上扩展转接逻辑
- ❌ 不存在 → 需先创建该技能或调整方案

---

### 2.2 人员 contactId 确认（必须）

**任务：** 确认全科医生和运营人员的 contactId

**执行命令：**
```bash
# 检查 user_index.json 中是否有 contactId
cat /home/admin/.openclaw/workspace/user_index.json | jq '.[] | select(.name == "郑小丹" or .name == "黄达" or .name == "高巧津" or .name == "翁锦秀" or .name == "陈能" or .name == "陈雅希")'
```

**处理方式：**
- ✅ 有 contactId → 直接写入配置文件
- ❌ 无 contactId → 联系对应人员生成 contactId 后补充

---

## 三、配置文件设计

### 3.1 文件结构

```
instance/config/
├── operators.json              # 运营人员配置（已有，扩展字段）
├── gp-doctors.json             # 全科医生配置（新增）
├── notify-templates.json       # 通知模板配置（新增）
└── escalation-policy.json      # 转接策略配置（新增）
```

---

### 3.2 config/operators.json（扩展）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/operators.json`

**内容：**
```json
{
  "name": "小医项目运营人员配置",
  "description": "无法解决的问题转接给以下运营人员",
  "updatedAt": "2026-04-29",
  "updatedBy": "高祖峰",
  "operators": [
    {
      "name": "翁锦秀",
      "phone": "13625049017",
      "contactId": "",
      "title": "运营主管",
      "department": "无边界运营平台",
      "status": "active",
      "addedAt": "2026-04-20",
      "notifyPriority": 1,
      "workingHours": "9:00-18:00"
    },
    {
      "name": "陈能",
      "phone": "18259897926",
      "contactId": "ky7794pk",
      "title": "主管",
      "department": "无边界运营平台 > 内容运营组 > 医学内容运营线",
      "status": "active",
      "addedAt": "2026-04-20",
      "notifyPriority": 2,
      "workingHours": "9:00-18:00"
    },
    {
      "name": "陈雅希",
      "phone": "15859099730",
      "contactId": "qenn2znq",
      "title": "助理产品经理",
      "department": "无边界产品平台",
      "status": "active",
      "addedAt": "2026-04-21",
      "notifyPriority": 99,
      "role": "fallback",
      "workingHours": "9:00-18:00"
    }
  ]
}
```

**字段说明：**
| 字段 | 必填 | 说明 |
|------|------|------|
| name | ✅ | 姓名 |
| phone | ✅ | 手机号（脱敏存储） |
| contactId | ⚠️ | 用于发送消息，如为空需后续补充 |
| status | ✅ | active/inactive |
| notifyPriority | ✅ | 通知优先级（数字越小优先级越高） |
| role | ❌ | fallback=兜底联系人（不填则正常轮询） |
| workingHours | ❌ | 工作时间，用于用户告知模板 |

---

### 3.3 config/gp-doctors.json（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/gp-doctors.json`

**内容：**
```json
{
  "name": "全科医生团队配置",
  "description": "审核催促工单转接对象",
  "updatedAt": "2026-04-29",
  "updatedBy": "高祖峰",
  "doctors": [
    {
      "name": "郑小丹",
      "phone": "13515003820",
      "contactId": "qrryywpq",
      "title": "全科医生",
      "department": "无边界医生平台 > 全职医生组",
      "status": "active",
      "addedAt": "2026-04-27",
      "notifyPriority": 1
    },
    {
      "name": "黄达",
      "phone": "13705949274",
      "contactId": "q49pp4zk",
      "title": "全科医生",
      "department": "无边界医生平台 > 全职医生组",
      "status": "active",
      "addedAt": "2026-04-27",
      "notifyPriority": 2
    },
    {
      "name": "高巧津",
      "phone": "13959111325",
      "contactId": "v18m32zk",
      "title": "全科医生",
      "department": "无边界医生平台 > 全职医生组",
      "status": "active",
      "addedAt": "2026-04-27",
      "notifyPriority": 3
    }
  ],
  "notifyPolicy": {
    "method": "round_robin",
    "description": "轮询通知，避免重复打扰同一医生",
    "retryCount": 3,
    "retryIntervalSeconds": 300
  }
}
```

**字段说明：**
| 字段 | 必填 | 说明 |
|------|------|------|
| name | ✅ | 姓名 |
| phone | ✅ | 手机号（脱敏存储） |
| contactId | ⚠️ | 用于发送消息 |
| status | ✅ | active/inactive |
| notifyPriority | ✅ | 通知优先级（数字越小优先级越高） |

---

### 3.4 config/notify-templates.json（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/notify-templates.json`

**内容：**
```json
{
  "description": "智能转接通知模板配置",
  "updatedAt": "2026-04-29",
  "updatedBy": "高祖峰",
  "templates": {
    "operator-escalation": {
      "description": "运营人员转接通知（发送给运营人员）",
      "content": "📋【问题转接通知】\n\n用户信息：\n• 姓名：{name}\n• 手机号：{phone}\n• contactId: {contactId}\n\n问题描述：\n{problem}\n\n转接时间：{notify_time}\n工单 ID: {ticket_id}\n\n请专人与用户联系处理，谢谢！"
    },
    "gp-doctor-escalation": {
      "description": "全科医生审核催促（发送给全科医生）",
      "content": "🔔【审核催促工单】\n\n用户信息：\n• 姓名：{name}\n• 手机号：{phone}\n• 医生 UID: {doctor_uid}\n\n待审核详情：\n• 活动：{activity}\n• 业务类型：{business_type}\n• 数量：{count}\n• 提交时间：{submit_time}\n\n请全科医生团队尽快审核，谢谢！\n\n工单 ID: {ticket_id}\n时间：{notify_time}"
    },
    "user-escalation-notify": {
      "description": "用户告知模板（问题转接）",
      "content": "您好，您的问题我已记录，将转接给专属运营人员处理。\n\n运营人员将在工作时间（{working_hours}）主动与您联系，请耐心等待。\n\n如长时间未收到回复，您可再次联系我查询进度。\n\n工单号：{ticket_id}"
    },
    "user-audit-escalation-notify": {
      "description": "用户告知模板（审核催促）",
      "content": "您好，查询到您的业务内容当前状态为【待审核】。\n\n📋 待审核详情：\n• 活动：{activity}\n• 业务类型：{business_type}\n• 数量：{count}\n\n🔄 已为您转接至内部全科医生团队催促审核。\n\n工作人员会尽快处理，审核完成后我会第一时间通知您！\n\n工单号：{ticket_id}"
    },
    "fallback-notify": {
      "description": "兜底通知（发送给陈雅希）",
      "content": "【小医转接异常通知】\n\n医生：{doctor_name}\n问题：{problem}\n转接失败原因：{failure_reason}\n时间：{notify_time}\n\n请及时处理，谢谢！"
    }
  },
  "variables": {
    "description": "模板变量说明",
    "common": {
      "name": "用户/医生姓名",
      "phone": "用户/医生手机号",
      "contactId": "用户 contactId",
      "notify_time": "通知时间（YYYY-MM-DD HH:mm）",
      "ticket_id": "工单 ID"
    },
    "operator-escalation": {
      "problem": "问题描述"
    },
    "gp-doctor-escalation": {
      "doctor_uid": "医生 UID",
      "activity": "活动名称",
      "business_type": "业务类型",
      "count": "数量",
      "submit_time": "提交时间"
    },
    "user-escalation-notify": {
      "working_hours": "运营人员工作时间（如 9:00-18:00）"
    },
    "fallback-notify": {
      "doctor_name": "医生姓名",
      "failure_reason": "转接失败原因"
    }
  }
}
```

---

### 3.5 config/escalation-policy.json（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/escalation-policy.json`

**内容：**
```json
{
  "description": "智能转接策略配置",
  "updatedAt": "2026-04-29",
  "updatedBy": "高祖峰",
  "policies": {
    "anti-repeat": {
      "enabled": true,
      "windowHours": 24,
      "description": "24 小时内同一用户同一问题只通知一次",
      "recordFile": "config/last_escalation_notify.json"
    },
    "round-robin": {
      "enabled": true,
      "method": "by_hour",
      "description": "按小时轮询通知，确保均匀分布",
      "algorithm": "hour % len(operators)"
    },
    "retry": {
      "enabled": true,
      "maxRetries": 3,
      "retryIntervalSeconds": 300,
      "description": "通知失败重试 3 次，间隔 5 分钟"
    },
    "fallback": {
      "enabled": true,
      "condition": "all_failed",
      "description": "所有运营人员通知失败时，通知兜底联系人"
    },
    "workingHours": {
      "enabled": false,
      "start": "9:00",
      "end": "18:00",
      "timezone": "Asia/Shanghai",
      "description": "工作时间通知（可选功能，默认关闭）"
    }
  }
}
```

---

## 四、代码实现方案

### 4.1 文件结构

```
instance/skills/ticket-escalation/
├── SKILL.md                          # 技能说明（修改）
├── scripts/
│   ├── smart_escalation.py           # 智能转接主脚本（新增）
│   └── __init__.py
└── utils/
    ├── config_loader.py              # 配置加载工具（新增）
    ├── template_engine.py            # 模板引擎（新增）
    └── notify_policy.py              # 通知策略（新增）
```

---

### 4.2 utils/config_loader.py（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/skills/ticket-escalation/utils/config_loader.py`

**内容：**
```python
#!/usr/bin/env python3
"""配置加载工具模块"""

import json
import os
from pathlib import Path
from typing import Dict, List, Any, Optional

# 配置目录（相对于 instance 目录）
CONFIG_DIR = Path(__file__).parent.parent.parent / "config"


class ConfigLoader:
    """配置加载器"""
    
    def __init__(self, config_dir: Optional[Path] = None):
        self.config_dir = config_dir or CONFIG_DIR
    
    def load_json(self, filename: str) -> Dict[str, Any]:
        """加载 JSON 配置文件"""
        filepath = self.config_dir / filename
        if not filepath.exists():
            raise FileNotFoundError(f"配置文件不存在：{filepath}")
        
        with open(filepath, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    def load_operators(self, active_only: bool = True) -> List[Dict[str, Any]]:
        """加载运营人员配置"""
        config = self.load_json("operators.json")
        operators = config.get("operators", [])
        
        if active_only:
            operators = [o for o in operators if o.get("status") == "active"]
        
        # 按优先级排序
        operators.sort(key=lambda x: x.get("notifyPriority", 99))
        return operators
    
    def load_gp_doctors(self, active_only: bool = True) -> List[Dict[str, Any]]:
        """加载全科医生配置"""
        config = self.load_json("gp-doctors.json")
        doctors = config.get("doctors", [])
        
        if active_only:
            doctors = [d for d in doctors if d.get("status") == "active"]
        
        # 按优先级排序
        doctors.sort(key=lambda x: x.get("notifyPriority", 99))
        return doctors
    
    def load_templates(self) -> Dict[str, str]:
        """加载通知模板"""
        config = self.load_json("notify-templates.json")
        templates = config.get("templates", {})
        return {name: t.get("content", "") for name, t in templates.items()}
    
    def load_template(self, template_name: str) -> str:
        """加载单个模板"""
        templates = self.load_templates()
        if template_name not in templates:
            raise KeyError(f"模板不存在：{template_name}")
        return templates[template_name]
    
    def load_policy(self) -> Dict[str, Any]:
        """加载转接策略配置"""
        config = self.load_json("escalation-policy.json")
        return config.get("policies", {})
    
    def get_fallback_contact(self) -> Optional[Dict[str, Any]]:
        """获取兜底联系人"""
        operators = self.load_operators(active_only=True)
        fallbacks = [o for o in operators if o.get("role") == "fallback"]
        return fallbacks[0] if fallbacks else None


# 单例实例
_config_loader = None

def get_config_loader() -> ConfigLoader:
    """获取配置加载器单例"""
    global _config_loader
    if _config_loader is None:
        _config_loader = ConfigLoader()
    return _config_loader
```

---

### 4.3 utils/template_engine.py（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/skills/ticket-escalation/utils/template_engine.py`

**内容：**
```python
#!/usr/bin/env python3
"""模板引擎模块"""

from datetime import datetime
from typing import Dict, Any


class TemplateEngine:
    """模板引擎"""
    
    @staticmethod
    def format(template: str, variables: Dict[str, Any]) -> str:
        """格式化模板"""
        # 添加默认变量
        variables.setdefault('notify_time', datetime.now().strftime('%Y-%m-%d %H:%M'))
        
        # 安全格式化（缺失变量时保留原样）
        try:
            return template.format(**variables)
        except KeyError as e:
            # 记录日志并返回原始模板
            print(f"⚠️ 模板变量缺失：{e}")
            return template
    
    @staticmethod
    def validate_template(template: str) -> tuple[bool, list]:
        """验证模板格式"""
        import re
        # 查找所有 {variable} 格式的占位符
        placeholders = re.findall(r'\{(\w+)\}', template)
        
        # 检查是否有未闭合的括号
        if template.count('{') != template.count('}'):
            return False, ["模板括号不匹配"]
        
        # 检查占位符命名
        invalid = [p for p in placeholders if not p.isidentifier()]
        if invalid:
            return False, [f"无效的占位符：{invalid}"]
        
        return True, placeholders


def format_template(template_name: str, variables: Dict[str, Any]) -> str:
    """便捷函数：加载并格式化模板"""
    from .config_loader import get_config_loader
    
    loader = get_config_loader()
    template = loader.load_template(template_name)
    return TemplateEngine.format(template, variables)
```

---

### 4.4 utils/notify_policy.py（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/skills/ticket-escalation/utils/notify_policy.py`

**内容：**
```python
#!/usr/bin/env python3
"""通知策略模块"""

import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Any, Optional

# 防重复记录文件
LAST_NOTIFY_FILE = Path(__file__).parent.parent.parent / "config" / "last_escalation_notify.json"


class NotifyPolicy:
    """通知策略"""
    
    def __init__(self, policy_config: Dict[str, Any]):
        self.config = policy_config
        self.anti_repeat = policy_config.get("anti-repeat", {})
        self.round_robin = policy_config.get("round-robin", {})
        self.retry = policy_config.get("retry", {})
    
    def should_notify(self, ticket_type: str, user_id: str) -> bool:
        """检查是否应该通知（防重复机制）"""
        if not self.anti_repeat.get("enabled", True):
            return True
        
        window_hours = self.anti_repeat.get("windowHours", 24)
        key = f"{ticket_type}:{user_id}"
        
        # 加载历史记录
        history = self._load_history()
        last_time = history.get(key)
        
        if last_time:
            last_dt = datetime.fromisoformat(last_time)
            if datetime.now() - last_dt < timedelta(hours=window_hours):
                return False
        
        return True
    
    def record_notify(self, ticket_type: str, user_id: str):
        """记录通知时间"""
        key = f"{ticket_type}:{user_id}"
        history = self._load_history()
        history[key] = datetime.now().isoformat()
        self._save_history(history)
    
    def select_recipient(self, recipients: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
        """选择接收人（轮询算法）"""
        if not recipients:
            return None
        
        method = self.round_robin.get("method", "by_hour")
        
        if method == "by_hour":
            # 按小时轮询
            index = datetime.now().hour % len(recipients)
            return recipients[index]
        elif method == "by_day":
            # 按天轮询
            index = datetime.now().timetuple().tm_yday % len(recipients)
            return recipients[index]
        else:
            # 默认返回第一个
            return recipients[0]
    
    def get_retry_config(self) -> tuple[int, int]:
        """获取重试配置"""
        max_retries = self.retry.get("maxRetries", 3)
        interval = self.retry.get("retryIntervalSeconds", 300)
        return max_retries, interval
    
    def _load_history(self) -> Dict[str, str]:
        """加载通知历史记录"""
        if not LAST_NOTIFY_FILE.exists():
            return {}
        
        try:
            with open(LAST_NOTIFY_FILE, 'r', encoding='utf-8') as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError):
            return {}
    
    def _save_history(self, history: Dict[str, str]):
        """保存通知历史记录"""
        LAST_NOTIFY_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(LAST_NOTIFY_FILE, 'w', encoding='utf-8') as f:
            json.dump(history, f, indent=2, ensure_ascii=False)
```

---

### 4.5 scripts/smart_escalation.py（新增）

**文件路径：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/skills/ticket-escalation/scripts/smart_escalation.py`

**内容：**
```python
#!/usr/bin/env python3
"""智能转接脚本"""

import json
import subprocess
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, Optional

# 导入工具模块
from ..utils.config_loader import get_config_loader
from ..utils.template_engine import format_template
from ..utils.notify_policy import NotifyPolicy

# 工单目录
TICKETS_DIR = Path(__file__).parent.parent.parent / "tickets"


class SmartEscalation:
    """智能转接处理器"""
    
    def __init__(self):
        self.config_loader = get_config_loader()
        self.policy = NotifyPolicy(self.config_loader.load_policy())
    
    def create_ticket(self, user_info: Dict[str, Any], problem: str, 
                      assigned_to: str, ticket_type: str) -> str:
        """创建工单"""
        TICKETS_DIR.mkdir(parents=True, exist_ok=True)
        
        # 生成工单 ID
        date_str = datetime.now().strftime('%Y%m%d')
        seq = self._get_next_seq(date_str)
        ticket_id = f"ticket-{date_str}-{seq:03d}"
        
        # 创建工单数据
        ticket = {
            "ticketId": ticket_id,
            "type": ticket_type,
            "status": "pending",
            "createdAt": datetime.now().isoformat(),
            "user": user_info,
            "problem": problem,
            "assignedTo": assigned_to,
            "resolvedAt": None,
            "notes": []
        }
        
        # 保存工单文件
        ticket_file = TICKETS_DIR / f"{ticket_id}.json"
        with open(ticket_file, 'w', encoding='utf-8') as f:
            json.dump(ticket, f, indent=2, ensure_ascii=False)
        
        return ticket_id
    
    def escalate_to_operator(self, user_info: Dict[str, Any], problem: str) -> Optional[str]:
        """转接至运营人员"""
        # 检查防重复
        user_id = user_info.get('contactId') or user_info.get('phone')
        if not self.policy.should_notify('operator', user_id):
            print(f"⚠️ 24 小时内已通知过，跳过")
            return None
        
        # 加载运营人员
        operators = self.config_loader.load_operators()
        operators = [o for o in operators if o.get('role') != 'fallback']
        
        if not operators:
            print("❌ 无可用运营人员")
            return None
        
        # 选择接收人
        selected = self.policy.select_recipient(operators)
        
        # 创建工单
        ticket_id = self.create_ticket(user_info, problem, selected['name'], 'operator-escalation')
        
        # 发送通知
        template = self.config_loader.load_template('operator-escalation')
        message = format_template(template, {
            'name': user_info.get('name', ''),
            'phone': user_info.get('phone', ''),
            'contactId': user_info.get('contactId', ''),
            'problem': problem,
            'ticket_id': ticket_id
        })
        
        # 调用 message 工具发送（需适配 OpenClaw message 工具）
        self._send_message(selected.get('contactId'), message)
        
        # 记录通知
        self.policy.record_notify('operator', user_id)
        
        print(f"✅ 已通知运营人员{selected['name']}，工单号：{ticket_id}")
        return ticket_id
    
    def escalate_to_gp_doctor(self, user_info: Dict[str, Any], 
                                audit_info: Dict[str, Any]) -> Optional[str]:
        """转接至全科医生（审核催促）"""
        # 检查防重复
        user_id = user_info.get('contactId') or user_info.get('phone')
        if not self.policy.should_notify('gp-doctor', user_id):
            print(f"⚠️ 24 小时内已通知过，跳过")
            return None
        
        # 加载全科医生
        doctors = self.config_loader.load_gp_doctors()
        
        if not doctors:
            print("❌ 无可用全科医生")
            return None
        
        # 选择接收人
        selected = self.policy.select_recipient(doctors)
        
        # 创建工单
        ticket_id = self.create_ticket(user_info, str(audit_info), selected['name'], 'audit-escalation')
        
        # 发送通知
        template = self.config_loader.load_template('gp-doctor-escalation')
        message = format_template(template, {
            'name': user_info.get('name', ''),
            'phone': user_info.get('phone', ''),
            'doctor_uid': audit_info.get('doctorUid', ''),
            'activity': audit_info.get('activity', ''),
            'business_type': audit_info.get('businessType', ''),
            'count': audit_info.get('count', ''),
            'submit_time': audit_info.get('submitTime', ''),
            'ticket_id': ticket_id
        })
        
        # 调用 message 工具发送
        self._send_message(selected.get('contactId'), message)
        
        # 记录通知
        self.policy.record_notify('gp-doctor', user_id)
        
        print(f"✅ 已通知全科医生{selected['name']}，工单号：{ticket_id}")
        return ticket_id
    
    def notify_user(self, user_info: Dict[str, Any], ticket_id: str, 
                    notify_type: str = 'escalation') -> str:
        """通知用户已转接"""
        if notify_type == 'escalation':
            template = self.config_loader.load_template('user-escalation-notify')
            # 获取工作时间
            operators = self.config_loader.load_operators()
            working_hours = operators[0].get('workingHours', '9:00-18:00') if operators else '9:00-18:00'
            message = format_template(template, {
                'working_hours': working_hours,
                'ticket_id': ticket_id
            })
        elif notify_type == 'audit-escalation':
            template = self.config_loader.load_template('user-audit-escalation-notify')
            message = format_template(template, {
                'activity': user_info.get('activity', ''),
                'business_type': user_info.get('businessType', ''),
                'count': user_info.get('count', ''),
                'ticket_id': ticket_id
            })
        else:
            message = f"工单号：{ticket_id}"
        
        return message
    
    def _send_message(self, contact_id: str, message: str):
        """发送消息（调用 OpenClaw message 工具）"""
        if not contact_id:
            print("⚠️ contactId 为空，无法发送消息")
            return
        
        # 调用 OpenClaw message 工具
        # 注意：实际使用时需根据 OpenClaw message 工具 API 调整
        cmd = [
            'openclaw', 'message', 'send',
            '--channel', 'tutu-aggchat',
            '--target', contact_id,
            '--message', message
        ]
        
        try:
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            if result.returncode != 0:
                print(f"❌ 发送失败：{result.stderr}")
        except subprocess.TimeoutExpired:
            print("❌ 发送超时")
        except Exception as e:
            print(f"❌ 发送异常：{e}")
    
    def _get_next_seq(self, date_str: str) -> int:
        """获取工单序号"""
        existing = list(TICKETS_DIR.glob(f"ticket-{date_str}-*.json"))
        if not existing:
            return 1
        
        # 提取最大序号
        max_seq = 0
        for f in existing:
            try:
                seq = int(f.stem.split('-')[-1])
                max_seq = max(max_seq, seq)
            except (ValueError, IndexError):
                continue
        
        return max_seq + 1


def main():
    """主函数（命令行调用）"""
    import sys
    
    if len(sys.argv) < 2:
        print("用法：python smart_escalation.py <operator|gp-doctor> [参数...]")
        sys.exit(1)
    
    escalation = SmartEscalation()
    cmd = sys.argv[1]
    
    if cmd == 'operator':
        # 测试运营人员转接
        user_info = {'name': '测试用户', 'phone': '13800138000', 'contactId': 'test123'}
        problem = '测试问题'
        ticket_id = escalation.escalate_to_operator(user_info, problem)
        if ticket_id:
            msg = escalation.notify_user(user_info, ticket_id, 'escalation')
            print(f"用户通知：{msg}")
    
    elif cmd == 'gp-doctor':
        # 测试全科医生转接
        user_info = {'name': '测试医生', 'phone': '13900139000', 'contactId': 'test456'}
        audit_info = {
            'doctorUid': 123456,
            'activity': '测试活动',
            'businessType': '视频答题',
            'count': 10,
            'submitTime': '2026-04-29 10:00'
        }
        ticket_id = escalation.escalate_to_gp_doctor(user_info, audit_info)
        if ticket_id:
            msg = escalation.notify_user(user_info, ticket_id, 'audit-escalation')
            print(f"用户通知：{msg}")
    
    else:
        print(f"未知命令：{cmd}")
        sys.exit(1)


if __name__ == '__main__':
    main()
```

---

## 五、执行步骤

### 步骤 10.1：确认 activity-content-audit-query 技能

**任务：** 确认技能是否存在

**执行命令：**
```bash
find /home/admin/.openclaw/workspace -type d -name "activity-content-audit-query"
```

**预期结果：**
- ✅ 存在 → 继续步骤 10.2
- ❌ 不存在 → 需先创建该技能或调整方案

**预计时间：** 5 分钟

---

### 步骤 10.2：扩展 config/operators.json

**任务：** 扩展运营人员配置，增加 notifyPriority 字段

**文件：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/operators.json`

**验收：** JSON 格式验证通过

**预计时间：** 10 分钟

---

### 步骤 10.3：创建 config/gp-doctors.json

**任务：** 创建全科医生团队配置

**文件：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/gp-doctors.json`

**验收：** JSON 格式验证通过

**预计时间：** 10 分钟

---

### 步骤 10.4：创建 config/notify-templates.json

**任务：** 创建通知模板配置

**文件：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/notify-templates.json`

**验收：** 5 种模板完整，格式正确

**预计时间：** 15 分钟

---

### 步骤 10.5：创建 config/escalation-policy.json

**任务：** 创建转接策略配置

**文件：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/config/escalation-policy.json`

**验收：** JSON 格式验证通过

**预计时间：** 10 分钟

---

### 步骤 10.6：创建工具模块

**任务：** 创建配置加载、模板引擎、通知策略工具

**文件：**
- `utils/config_loader.py`
- `utils/template_engine.py`
- `utils/notify_policy.py`

**验收：** 模块可导入，无语法错误

**预计时间：** 40 分钟

---

### 步骤 10.7：创建智能转接脚本

**任务：** 创建 smart_escalation.py 主脚本

**文件：** `scripts/smart_escalation.py`

**验收：** 脚本可执行，无语法错误

**预计时间：** 40 分钟

---

### 步骤 10.8：修改 ticket-escalation/SKILL.md

**任务：** 增加自动触发逻辑说明

**文件：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/skills/ticket-escalation/SKILL.md`

**验收：** 技能说明完整

**预计时间：** 15 分钟

---

### 步骤 10.9：修改 AGENTS.md

**任务：** 新增智能转接机制说明

**文件：** `/home/admin/.openclaw/workspace/projects/xiaoyi-assistant/instance/AGENTS.md`

**验收：** 文档更新完成

**预计时间：** 15 分钟

---

### 步骤 10.10：测试验证

**任务：** 测试配置读取、通知发送、防重复机制

**测试用例：**
1. 配置加载测试
2. 模板格式化测试
3. 防重复通知测试（连续执行 2 次）
4. 轮询通知测试（连续执行多次）

**验收：** 所有测试通过

**预计时间：** 30 分钟

---

## 六、验收标准

| 验收项 | 标准 | 验证方法 |
|--------|------|----------|
| **配置创建** | 4 个配置文件格式正确 | jq 验证 JSON |
| **工具模块** | 3 个工具模块可导入 | Python import 测试 |
| **脚本执行** | smart_escalation.py 正常运行 | 手动执行测试 |
| **工单创建** | 正确创建工单文件 | 检查 tickets/目录 |
| **防重复** | 24 小时内不重复通知 | 连续执行 2 次验证 |
| **轮询** | 通知均匀分布 | 连续执行多次验证 |
| **模板** | 5 种模板完整 | 人工检查 |

---

## 七、风险评估

| 风险 | 可能性 | 影响 | 缓解措施 |
|------|--------|------|----------|
| **技能不存在** | 🔴 高 | 高 | 步骤 10.1 先确认 |
| **contactId 缺失** | 🟡 中 | 中 | 联系人员生成后补充 |
| **通知 API 失败** | 🟡 中 | 中 | 重试机制 + 兜底通知 |
| **配置格式错误** | 🟢 低 | 低 | JSON 验证 + 测试 |

---

## 八、回滚方案

**如执行中出现问题：**

1. **备份恢复：**
   ```bash
   # 备份配置文件
   cp config/operators.json config/operators.json.bak
   ```

2. **代码回滚：**
   ```bash
   # 删除新增文件
   rm -rf skills/ticket-escalation/scripts/
   rm -rf skills/ticket-escalation/utils/
   ```

3. **文档恢复：**
   ```bash
   # 恢复 AGENTS.md
   git checkout AGENTS.md
   ```

---

## 九、执行时间汇总

| 步骤 | 任务 | 预计时间 |
|------|------|----------|
| 10.1 | 确认技能 | 5 分钟 |
| 10.2 | 扩展 operators.json | 10 分钟 |
| 10.3 | 创建 gp-doctors.json | 10 分钟 |
| 10.4 | 创建 notify-templates.json | 15 分钟 |
| 10.5 | 创建 escalation-policy.json | 10 分钟 |
| 10.6 | 创建工具模块 | 40 分钟 |
| 10.7 | 创建智能转接脚本 | 40 分钟 |
| 10.8 | 修改 SKILL.md | 15 分钟 |
| 10.9 | 修改 AGENTS.md | 15 分钟 |
| 10.10 | 测试验证 | 30 分钟 |
| **合计** | | **3 小时 10 分钟** |

---

_执行计划版本：v3.0（配置化优化版）_  
_编制时间：2026-04-29 15:40_  
_状态：待确认执行_